package com.visec.hdy.hadoop.context;

import java.io.File;
import java.util.List;

import org.apache.commons.math3.exception.NumberIsTooSmallException;
import org.apache.log4j.Logger;

import com.visec.hdy.hadoop.standard.ReadContext;

/**
 * 基础读上下文实现类
 * 
 * @author hdy
 *
 * @param <K>
 */
public class LogReadContextImpl extends DefaultContextImpl implements ReadContext {
	protected List<File> files;
	protected int index;
	protected long endCurrent;
	protected long startCurrent;

	final Logger l = Logger.getLogger(LogReadContextImpl.class);

	public LogReadContextImpl(String nameSpace, long count) {
		super(nameSpace, count);
		endCurrent = count;
	}

	@Override
	public void back(long back) {
		if (files == null)
			return;
		if (back < 0)
			try {
				throw new NumberIsTooSmallException(back, 0, false);
			} catch (Exception e) {
				l.error("", e);
				return;
			}
		if (back > (index * count + current - startCurrent)) {
			index = 0;
			current = startCurrent;
			return;
		}
		index = index - (int) (back / count);
		current = current - (back % count);
		if (current < 0) {
			current = current + count;
			index--;
		}
	}

	@Override
	public void seek(long l) {
		current = startCurrent;
		index = 0;
		skip(l);
	}

	@Override
	public void skip(long skip) {
		if (files == null)
			return;
		if (skip < 0)
			try {
				throw new NumberIsTooSmallException(skip, 0, false);
			} catch (Exception e) {
				l.error("", e);
				return;
			}
		if ((skip + index * count + current) > (count * (files.size() - 1) + endCurrent)) {
			index = files.size() - 1;
			current = endCurrent;
			return;
		}
		index = index + (int) (skip / count);
		current = current + (skip % count);
		if (current > count) {
			current = current - count;
			index++;
		}
	}

	@Override
	public long read(long l) {
		if (l < 0)
			return -1;

		long oldCurrent = current;
		int oldIndex = index;

		if (current == endCurrent && index == (files.size() - 1))
			return -1;

		if (current == count) {
			index++;
			current = 0;
		}

		readMove(l);

		return index * count + current - (oldIndex * count) - oldCurrent;
	}

	/**
	 * 移动读取位置
	 * 
	 * @param l
	 */
	protected void readMove(long l) {
		if (index == (files.size() - 1)) {
			current = (current + l) > endCurrent ? endCurrent : current + l;
		} else if ((current + l) > count) {
			current = count;
		} else {
			current += l;
		}
	}

	@Override
	public File getFile() {
		return files.get(index);
	}

	@Override
	public void setQueryFile(List<File> files) {
		seek(0);
		this.files = sortFiles(files);
	}

	@Override
	public void setEndCurrent(long end) {
		if (end > count)
			try {
				throw new NumberIsTooSmallException(end, 0, false);
			} catch (Exception e) {
				l.error("", e);
				return;
			}
		this.endCurrent = end;
	}

	@Override
	public void setStartCurrent(long start) {
		if (start > count)
			try {
				throw new NumberIsTooSmallException(start, 0, false);
			} catch (Exception e) {
				l.error("", e);
				return;
			}
		this.startCurrent = start;
		this.current = start;
	}

	@Override
	public long dataCount() {
		return count * (files.size() - 1) + endCurrent - startCurrent;
	}
}
